package com.idanchuang.count.listener;

import com.alibaba.fastjson.JSON;
import com.idanchuang.count.core.domain.AnalyzeCounter;
import com.idanchuang.count.core.entity.SimpleCountIncrEvent;
import com.idanchuang.count.core.model.BizConfig;
import com.idanchuang.count.core.model.BizProperties;
import com.idanchuang.count.dao.CounterAnalyzeDao;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 监听 base_count.* 的topic，然后批量写入 tdEngine
 * @author Richard_yyf
 * @version 1.0 2021/10/12
 */
@Slf4j
public class SimpleCountAnalyzeListener {

    private final CounterAnalyzeDao counterAnalyzeDao;

    @Resource
    private BizProperties bizProperties;

    public SimpleCountAnalyzeListener(CounterAnalyzeDao counterAnalyzeDao) {
        this.counterAnalyzeDao = counterAnalyzeDao;
    }

    @KafkaListener(
            topicPattern = "base_count.*",
//            topics = {"base_count_single_test", "base_count_batch_test"}, // for test
            containerFactory = "simpleCountAnalyzeBatchConsumeContainerFactory",
            clientIdPrefix = "count-analyze"
    )
    public void batchConsumeIncrEventForAnalyze(@Payload(required = false) List<SimpleCountIncrEvent> incrEvents,
                                      Acknowledgment ack) {

        Map<String, BizConfig> bizConfigMap = bizProperties.getConfig();
        List<AnalyzeCounter> counterRecords = new ArrayList<>(incrEvents.size());
        for (SimpleCountIncrEvent incrEvent : incrEvents) {
            if (incrEvent == null) {
                continue;
            }
            if (incrEvent.getTs() == null) {
                log.warn("ts get null, skip. 内容: {}", incrEvent);
                continue;
            }
            if (!bizConfigMap.containsKey(incrEvent.getBizCode())
                    || !bizConfigMap.get(incrEvent.getBizCode()).isCountAnalyzeEnable()) {
                // bizCode 未配置 或者 CountAnalyzeEnable false
                log.info("illegal bizCode[{}] for CountAnalyze", incrEvent.getBizCode());
                continue;
            }

            try {
                counterRecords.add(AnalyzeCounter.buildFrom(incrEvent));
            } catch (Exception e) {
                // just log and ignore
                log.error("buildAnalyzeCounter error, incrEvent: " + JSON.toJSONString(incrEvent), e);
            }
        }

        if (counterRecords.size() == 0) {
            ack.acknowledge();
            return;
        }

        // tdEngine insert
        try {
            int result = counterAnalyzeDao.insert2TablesAndAutoCreateTableWhenNotExist(counterRecords);
            log.info("tdEngine insert, input size: {}, success: {}", counterRecords.size(), result);
        } catch (Exception e) {
            log.error("tdEngine batch insert error", e);
            throw e;
        }

        ack.acknowledge();
    }
}
